use std::{collections::BTreeMap, ops::Range};

use common_base::{
    bytes::{BufMut, Bytes, BytesMut},
    prelude::*,
    serde::{
        ser::{Error, SerializeStruct},
        Serialize, Serializer,
    },
    serde_json,
    sqlx::{mysql::MySqlRow, Column, Row},
};

use crate::data::{
    dataset::{data_status::DataStatus, data_value::ValueType},
    DataValue,
};

#[derive(Debug)]
pub struct Rowset {
    storage: Bytes,
    column_position_map: BTreeMap<String, usize>,
    value_ranges: Vec<Option<Range<usize>>>,
    value_types: Vec<ValueType>,
    data_status: DataStatus,
}

impl Default for Rowset {
    fn default() -> Self {
        Self {
            storage: Default::default(),
            column_position_map: Default::default(),
            value_ranges: Default::default(),
            value_types: Default::default(),
            data_status: DataStatus::Normal,
        }
    }
}

impl TryFrom<MySqlRow> for Rowset {
    type Error = ErrorCode;

    fn try_from(db_row: MySqlRow) -> std::result::Result<Self, Self::Error> {
        let mut rowset_builder = RowsetBuilder::default();
        for col in db_row.columns() {
            let value = db_row.try_get_raw(col.ordinal())?;

            let data_value: DataValue = (col.ordinal(), value).try_into()?;
            rowset_builder.put_data_value(col.name(), data_value);
        }
        let mut rowset: Rowset = rowset_builder.freeze();
        rowset.data_status = DataStatus::Normal;
        Ok(rowset)
    }
}

impl Serialize for Rowset {
    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let mut rowset = serializer.serialize_struct("RowSet", 2)?;

        let mut data_values: BTreeMap<String, serde_json::Value> = BTreeMap::new();
        for (col_name, col_pos) in self.column_position_map.iter() {
            let value_type = self.value_types[*col_pos];
            if let ValueType::Null = value_type {
                data_values.insert(col_name.to_owned(), serde_json::Value::Null);
                continue;
            }

            let range = self.value_ranges[*col_pos]
                .as_ref()
                .ok_or(ErrorCode::UnImplement(format!(
                    "can not find range for {} at {}",
                    col_name, col_pos
                )))
                .map_err(S::Error::custom)?;
            let bytes = self
                .storage
                .get(range.start..range.end)
                .ok_or(ErrorCode::UnImplement(format!(
                    "can not find bytes for {} at {} [{}..{}]",
                    col_name, col_pos, range.start, range.end
                )))
                .map_err(S::Error::custom)?;

            match value_type {
                ValueType::Null => {
                    data_values.insert(col_name.to_owned(), serde_json::Value::Null);
                }
                ValueType::Bool => {
                    let b = DataValue::parse_bool_from(bytes).map_err(S::Error::custom)?;
                    data_values.insert(col_name.to_owned(), serde_json::Value::Bool(b));
                }
                ValueType::U8 => {
                    if bytes.len() != 1 {
                        return Err(S::Error::custom(ErrorCode::UnImplement(format!(
                            "can not parse u8 {:?}",
                            bytes
                        ))));
                    }
                    data_values.insert(
                        col_name.to_owned(),
                        serde_json::Value::Number(bytes[0].into()),
                    );
                }
                ValueType::U16 => {
                    if bytes.len() != 2 {
                        return Err(S::Error::custom(ErrorCode::UnImplement(format!(
                            "can not parse u16 {:?}",
                            bytes
                        ))));
                    }
                    let u = u16::from_be_bytes(bytes.split_at(2).0.try_into().unwrap());
                    data_values.insert(col_name.to_owned(), serde_json::Value::Number(u.into()));
                }
                ValueType::U32 => {
                    if bytes.len() != 4 {
                        return Err(S::Error::custom(ErrorCode::UnImplement(format!(
                            "can not parse u32 {:?}",
                            bytes
                        ))));
                    }
                    let u = u32::from_be_bytes(bytes.split_at(4).0.try_into().unwrap());
                    data_values.insert(col_name.to_owned(), serde_json::Value::Number(u.into()));
                }
                ValueType::U64 => {
                    if bytes.len() != 8 {
                        return Err(S::Error::custom(ErrorCode::UnImplement(format!(
                            "can not parse u64 {:?}",
                            bytes
                        ))));
                    }
                    let u = u64::from_be_bytes(bytes.split_at(8).0.try_into().unwrap());
                    data_values.insert(col_name.to_owned(), serde_json::Value::Number(u.into()));
                }
                ValueType::I8 => {
                    if bytes.len() != 1 {
                        return Err(S::Error::custom(ErrorCode::UnImplement(format!(
                            "can not parse i8 {:?}",
                            bytes
                        ))));
                    }
                    let i = bytes[0] as i8;
                    data_values.insert(col_name.to_owned(), serde_json::Value::Number(i.into()));
                }
                ValueType::I16 => {
                    if bytes.len() != 2 {
                        return Err(S::Error::custom(ErrorCode::UnImplement(format!(
                            "can not parse i16 {:?}",
                            bytes
                        ))));
                    }
                    let i = i16::from_be_bytes(bytes.split_at(2).0.try_into().unwrap());
                    data_values.insert(col_name.to_owned(), serde_json::Value::Number(i.into()));
                }
                ValueType::I32 => {
                    if bytes.len() != 4 {
                        return Err(S::Error::custom(ErrorCode::UnImplement(format!(
                            "can not parse i32 {:?}",
                            bytes
                        ))));
                    }
                    let i = i32::from_be_bytes(bytes.split_at(4).0.try_into().unwrap());
                    data_values.insert(col_name.to_owned(), serde_json::Value::Number(i.into()));
                }
                ValueType::I64 => {
                    if bytes.len() != 8 {
                        return Err(S::Error::custom(ErrorCode::UnImplement(format!(
                            "can not parse i64 {:?}",
                            bytes
                        ))));
                    }
                    let i = i64::from_be_bytes(bytes.split_at(8).0.try_into().unwrap());
                    data_values.insert(col_name.to_owned(), serde_json::Value::Number(i.into()));
                }
                ValueType::F32 => {
                    if bytes.len() != 4 {
                        return Err(S::Error::custom(ErrorCode::UnImplement(format!(
                            "can not parse f32 {:?}",
                            bytes
                        ))));
                    }
                    let f = f32::from_be_bytes(bytes.split_at(4).0.try_into().unwrap()) as f64;
                    data_values.insert(
                        col_name.to_owned(),
                        serde_json::Value::Number(
                            serde_json::Number::from_f64(f)
                                .unwrap_or(serde_json::Number::from_f64(0.0f64).unwrap()),
                        ),
                    );
                }
                ValueType::F64 => {
                    if bytes.len() != 8 {
                        return Err(S::Error::custom(ErrorCode::UnImplement(format!(
                            "can not parse f64 {:?}",
                            bytes
                        ))));
                    }
                    let f = f64::from_be_bytes(bytes.split_at(4).0.try_into().unwrap());
                    data_values.insert(
                        col_name.to_owned(),
                        serde_json::Value::Number(
                            serde_json::Number::from_f64(f)
                                .unwrap_or(serde_json::Number::from_f64(0.0f64).unwrap()),
                        ),
                    );
                }
                ValueType::BigDecimal
                | ValueType::String
                | ValueType::DateTime
                | ValueType::NaiveDate
                | ValueType::NaiveTime
                | ValueType::NaiveDateTime => {
                    data_values.insert(
                        col_name.to_owned(),
                        serde_json::Value::String(String::from_utf8_lossy(bytes).to_string()),
                    );
                }
            }
        }
        rowset.serialize_field("dataValues", &data_values)?;

        rowset.serialize_field("dataStatus", &self.data_status)?;

        rowset.end()
    }
}

struct RowsetBuilder {
    storage: BytesMut,
    value_ranges: Vec<Option<Range<usize>>>,
    value_types: Vec<ValueType>,
    column_position_max: usize,
    column_position_map: BTreeMap<String, usize>,
}

impl Default for RowsetBuilder {
    fn default() -> Self {
        Self {
            storage: Default::default(),
            value_ranges: Default::default(),
            value_types: Default::default(),
            column_position_max: 0,
            column_position_map: Default::default(),
        }
    }
}

impl RowsetBuilder {
    pub fn put_data_value(&mut self, col_name: &str, data_value: DataValue) {
        let current_pos = self.column_position_max;
        self.column_position_map
            .insert(col_name.to_owned(), current_pos);
        match data_value {
            DataValue::Null => {
                self.value_ranges.push(None);
                self.value_types.push(ValueType::Null);
            }
            DataValue::Bool(inner) => {
                let range = self.storage.len()..self.storage.len() + 1;
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::Bool);
                self.storage.put_u8(inner.into());
            }
            DataValue::U8(inner) => {
                let range = self.storage.len()..self.storage.len() + 1;
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::U8);
                self.storage.put_u8(inner);
            }
            DataValue::U16(inner) => {
                let range = self.storage.len()..self.storage.len() + 2;
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::U16);
                self.storage.put_u16(inner);
            }
            DataValue::U32(inner) => {
                let range = self.storage.len()..self.storage.len() + 4;
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::U32);
                self.storage.put_u32(inner);
            }
            DataValue::U64(inner) => {
                let range = self.storage.len()..self.storage.len() + 8;
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::U64);
                self.storage.put_u64(inner);
            }
            DataValue::I8(inner) => {
                let range = self.storage.len()..self.storage.len() + 1;
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::I8);
                self.storage.put_i8(inner);
            }
            DataValue::I16(inner) => {
                let range = self.storage.len()..self.storage.len() + 2;
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::I16);
                self.storage.put_i16(inner);
            }
            DataValue::I32(inner) => {
                let range = self.storage.len()..self.storage.len() + 4;
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::I32);
                self.storage.put_i32(inner);
            }
            DataValue::I64(inner) => {
                let range = self.storage.len()..self.storage.len() + 8;
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::I64);
                self.storage.put_i64(inner);
            }
            DataValue::F32(inner) => {
                let range = self.storage.len()..self.storage.len() + 4;
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::F32);
                self.storage.put_f32(inner);
            }
            DataValue::F64(inner) => {
                let range = self.storage.len()..self.storage.len() + 8;
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::F64);
                self.storage.put_f64(inner);
            }
            DataValue::BigDecimal(inner) => {
                let inner_bytes = inner.to_string();
                let range = self.storage.len()..self.storage.len() + inner_bytes.len();
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::BigDecimal);
                self.storage.put_slice(inner_bytes.as_bytes());
            }
            DataValue::String(inner) => {
                let range = self.storage.len()..self.storage.len() + inner.len();
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::String);
                self.storage.put_slice(inner.as_bytes());
            }
            DataValue::DateTime(inner) => {
                let inner_bytes = inner.to_string();
                let range = self.storage.len()..self.storage.len() + inner_bytes.len();
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::DateTime);
                self.storage.put_slice(inner_bytes.as_bytes());
            }
            DataValue::NaiveDate(inner) => {
                let inner_bytes = inner.to_string();
                let range = self.storage.len()..self.storage.len() + inner_bytes.len();
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::NaiveDate);
                self.storage.put_slice(inner_bytes.as_bytes());
            }
            DataValue::NaiveTime(inner) => {
                let inner_bytes = inner.to_string();
                let range = self.storage.len()..self.storage.len() + inner_bytes.len();
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::NaiveTime);
                self.storage.put_slice(inner_bytes.as_bytes());
            }
            DataValue::NaiveDateTime(inner) => {
                let inner_bytes = inner.to_string();
                let range = self.storage.len()..self.storage.len() + inner_bytes.len();
                self.value_ranges.push(Some(range));
                self.value_types.push(ValueType::NaiveDateTime);
                self.storage.put_slice(inner_bytes.as_bytes());
            }
        }
        self.column_position_max += 1;
    }

    fn freeze(self) -> Rowset {
        Rowset {
            data_status: DataStatus::Normal,
            column_position_map: self.column_position_map,
            storage: self.storage.freeze(),
            value_ranges: self.value_ranges,
            value_types: self.value_types,
        }
    }
}
